--首条轨迹后续更新即将超时
with line_info as (
    select waybill_no
         , send_station_network_code
         , send_station_network_name
    from (
             select waybill_no                                                             as waybill_no
                  , concat_ws(',', sort_array(collect_set(tbb.send_station_network_code))) as send_station_network_code
                  , concat_ws(',', sort_array(collect_set(tbb.send_station_network_name))) as send_station_network_name
             from (
                      select waybill_no
                           , concat(rn, ':', send_station_network_code) as send_station_network_code
                           , concat(rn, ':', send_station_network_name) as send_station_network_name
                      from jms_dm.dm_mng_thd_flase_send_waybill_detail_mid_dt
                      where dt >= date_add('{{ execution_date | cst_ds }}', -30)
                        and dt <= '{{ execution_date | cst_ds }}'
                  ) tbb
             group by tbb.waybill_no
         ) tt
),
     ship_upload_bill_tab as (
         select *
         from (
                  select bill_no
                       , upload_batch
                       , row_number() over (partition by bill_no order by create_time desc ) as rn1
                  from jms_dwd.dwd_tab_false_ship_upload_det_dt
                  where dt >= date_add('{{ execution_date | cst_ds }}', -30)
                    and dt <= date_add('{{ execution_date | cst_ds }}', 1)
                    and platform = '桃花岛春节不打烊'
              ) t
         where rn1 = 1
     ),
     mate_date as (
         select *
         from jms_dm.dm_mng_thd_flase_send_waybill_detail_mid_dt
         where dt >= date_add('{{ execution_date | cst_ds }}', -30)
           and dt <= '{{ execution_date | cst_ds }}'
     ),
     first_type as (
         select waybill_no
              , rn
              , dt
              , trace_falg
              , abs(round(duration, 2)) as duration
              , duty_network_code
              , duty_network_name
              , duty_standard
              , 1                       as duty_type
         from (
                  select waybill_no
                       , rn
                       , dt
--               , case when no_send_arr_flag = 0 then unix_timestamp(min_send_arr)-unix_timestamp(min_scan_time)
                       --1下一站为空,发件网点=揽收网点,且发件网点没有到件也没有发件,且没有下一站,(说明没有新轨迹,只有揽收操作),用当前时间-揽收时间
                       --2发件网点=揽收网点,且当前站点到件和发件只有一个,(说明没有新轨迹,只有一个操作) ,用到/发时间-揽收时间
                       --3下一站为空,且是始发网点/集散/中心只有(到or发),(说明没有下一轨迹),用当前时间-揽收时间
                       --4下一站为空,且始发集散/中心 有到且有发,用发件-到件
                       --4下一站不为空,用
                       , case
                             when send_station_network_code = taking_early_scansitecode and no_send_arr_flag = 1 and
                                  next_station_network_code is null
                                 then 1
                             when send_station_network_code = taking_early_scansitecode
                                 and (
                                          (send_station_send_later_time is null and
                                           send_station_arrival_early_time is not null) or
                                          (send_station_send_later_time is not null and
                                           send_station_arrival_early_time is null)
                                      )
                                 then 2
                             when next_station_network_code is null
                                 and (
                                          (send_station_send_later_time is null and
                                           send_station_arrival_early_time is not null) or
                                          (send_station_send_later_time is not null and
                                           send_station_arrival_early_time is null)
                                      )
                                 then 3
                             when next_station_network_code is null
                                 and (
                                      (send_station_send_later_time is not null and
                                       send_station_arrival_early_time is not null)
                                      )
                                 then 4
                             when next_station_network_code is not null
                                 and (
                                      (send_station_send_later_time is not null and
                                       send_station_arrival_early_time is not null)
                                      )
                                 then 5
                             when next_station_network_code is not null
                                 and (
                                      (next_station_send_later_time is not null or
                                       next_station_arrival_early_time is not null)
                                      )
                                 then 6
                             else 0 end            as trace_falg        --1,2无下一操作,3有下一操作 0其他情况应该没有
                       , case
                             when send_station_network_code = taking_early_scansitecode and no_send_arr_flag = 1 and
                                  next_station_network_code is null
                                 then (unix_timestamp(concat('{{ execution_date | cst_ds }}', ' 23:59:59')) -
                                       unix_timestamp(min_scan_time)) / 3600
                             when send_station_network_code = taking_early_scansitecode
                                 and (
                                          (send_station_send_later_time is null and
                                           send_station_arrival_early_time is not null) or
                                          (send_station_send_later_time is not null and
                                           send_station_arrival_early_time is null)
                                      )
                                 then (unix_timestamp(
                                               least(nvl(send_station_send_later_time, '9999-01-01 01:00:00'),
                                                     nvl(send_station_arrival_early_time, '9999-01-01 01:00:00'))) -
                                       unix_timestamp(min_scan_time)) / 3600
                             when next_station_network_code is null
                                 and (
                                          (send_station_send_later_time is null and
                                           send_station_arrival_early_time is not null) or
                                          (send_station_send_later_time is not null and
                                           send_station_arrival_early_time is null)
                                      )
                                 then (unix_timestamp(concat('{{ execution_date | cst_ds }}', ' 23:59:59')) -
                                       unix_timestamp(min_scan_time)) / 3600
                             when next_station_network_code is null
                                 and (
                                      (send_station_send_later_time is not null and
                                       send_station_arrival_early_time is not null)
                                      )
                                 then abs(unix_timestamp(send_station_send_later_time) -
                                          unix_timestamp(send_station_arrival_early_time)
                                          ) / 3600
                             when next_station_network_code is not null
                                 and (
                                      (send_station_send_later_time is not null and
                                       send_station_arrival_early_time is not null)
                                      )
                                 then abs(unix_timestamp(send_station_send_later_time) -
                                          unix_timestamp(send_station_arrival_early_time)
                                          ) / 3600
                             when next_station_network_code is not null
                                 and (
                                      (next_station_send_later_time is not null or
                                       next_station_arrival_early_time is not null)
                                      )
                                 then abs(unix_timestamp(least(nvl(next_station_send_later_time, '9999-01-01 01:00:00'),
                                                               nvl(next_station_arrival_early_time, '9999-01-01 01:00:00'))) -
                                          unix_timestamp(min_scan_time)
                                          ) / 3600
                             else 0 end            as duration          --环节时长
                       , send_station_network_code as duty_network_code --责任机构
                       , send_station_network_name as duty_network_name --责任机构
                       , 24                        as duty_standard
                  from mate_date
                  where rn = 1
              ) t
         where abs(duration) > duty_standard
     ),
--离开所在省份最后中心即将超时
     second_type as (
         --第一中心不为空,第二中心不为空,且同一省份
         --第一中心不为空,且有发件时间
         --第一中心不为空,且无发件时间,且存在第二中(发件,到件存在一个)
         --第一中心不为空,第二中心为空
         select waybill_no
              , rn
              , dt
              , trace_falg
              , abs(round(duration, 2)) as duration
              , duty_network_code
              , duty_network_name
              , duty_standard
              , 2                       as duty_type
         from (
                  select waybill_no
                       , rn
                       , dt
                       , case
                             when first_center_send_later_time is not null and
                                  second_center_send_later_time is not null and
                                  is_same_prov_flag = 1
                                 then 1
                             when first_center_send_later_time is not null
                                 then 2
                             when first_center_send_later_time is null and
                                  (second_center_arrival_early_time is not null or
                                   second_center_send_later_time is not null)
                                 then 3
                             when second_center_network_code is null
                                 then 4
                             else 0 end                                           as trace_falg
                       , case
                             when first_center_send_later_time is not null and
                                  second_center_send_later_time is not null and
                                  is_same_prov_flag = 1
                                 then (unix_timestamp(second_center_send_later_time) - unix_timestamp(min_scan_time)) /
                                      3600
                             when first_center_send_later_time is not null
                                 then (unix_timestamp(first_center_send_later_time) - unix_timestamp(min_scan_time)) /
                                      3600
                             when first_center_send_later_time is null and
                                  (second_center_arrival_early_time is not null or
                                   second_center_send_later_time is not null)
                                 then
                                     (unix_timestamp(coalesce(second_center_send_later_time,
                                                              second_center_arrival_early_time)) -
                                      unix_timestamp(min_scan_time)) / 3600
                             when second_center_network_code is null
                                 then (unix_timestamp(concat('{{ execution_date | cst_ds }}', ' 23:59:59')) -
                                       unix_timestamp(min_scan_time)) / 3600
                             else 0 end                                           as duration
--                        , case
--                              when first_center_network_code is not null and first_center_arrival_early_time is null
--                                  then first_center_network_code
--                              else send_station_network_code end                   as duty_network_code
                       , send_station_network_code                                as duty_network_code
                       , send_station_network_name                                as duty_network_name
                       , coalesce(extremely_remote_timeout, remote_timeout, '48') as duty_standard
                  from mate_date
                  where rn = 1
                    and first_center_network_code is not null and first_center_network_code <> ''
              ) t
         where abs(duration) > duty_standard
     ),
--同一中心停留即将超时
     third_type as (
         --中心有到,有发
         --中心有到,无发,且后续无轨迹
         select waybill_no
              , rn
              , dt
              , trace_falg
              , abs(round(duration, 2)) as duration
              , duty_network_code
              , duty_network_name
              , duty_standard
              , 3                       as duty_type
         from (
                  select waybill_no
                       , rn
                       , dt
                       , case
                             when send_station_send_later_time is not null and
                                  send_station_arrival_early_time is not null
                                 then 1
                             when (send_station_arrival_early_time is not null and
                                   send_station_send_later_time is null and next_station_network_code is null)
                                 then 2
                             else 0 end            as trace_falg
                       , case
                             when send_station_send_later_time is not null and
                                  send_station_arrival_early_time is not null
                                 then abs((unix_timestamp(send_station_send_later_time) -
                                           unix_timestamp(send_station_arrival_early_time))) / 3600
                             when (send_station_arrival_early_time is not null and
                                   send_station_send_later_time is null and next_station_network_code is null)
                                 then (unix_timestamp(concat('{{ execution_date | cst_ds }}', ' 23:59:59')) -
                                       unix_timestamp(send_station_arrival_early_time)) / 3600
                             else 0 end            as duration
                       , send_station_network_code as duty_network_code
                       , send_station_network_name as duty_network_name
                       , 24                        as duty_standard
                  from mate_date
                  where send_station_network_type = 2
              ) t
         where abs(duration) > duty_standard
     ),
--相邻物流轨迹更新即将超时
     four_type as (
         select waybill_no
              , rn
              , dt
              , trace_falg
              , abs(round(duration, 2)) as duration
              , duty_network_code
              , duty_network_name
              , duty_standard
              , 4                       as duty_type
         from (
                  select waybill_no
                       , rn
                       , dt
--                        , case
--                              when next_station_network_code is not null
--
--                                  then 1
--                              when next_station_network_code is null then 2
--                              else 0 end            as trace_falg
                       , case
                             when next_station_network_code is not null
                                 then 1
                             when next_station_network_code is null and deliver_early_scantime is null and
                                  sign_time is null
                                 then 2
                             when next_station_network_code is null and
                                  (deliver_early_scantime is not null or sign_time is not null)
                                 then 3
                             else 0 end            as trace_falg
                       , case
                             when next_station_network_code is not null and
                                  least(nvl(next_station_send_later_time, '9999-01-01 01:00:00'),
                                        nvl(next_station_arrival_early_time, '9999-01-01 01:00:00')) <
                                  nvl(deliver_early_scantime, '9999-01-01 01:00:00')
                                 then (unix_timestamp(least(nvl(next_station_send_later_time, '9999-01-01 01:00:00'),
                                                            nvl(next_station_arrival_early_time, '9999-01-01 01:00:00'))) -
                                       unix_timestamp(greatest(nvl(send_station_send_later_time, '0000-01-01 01:00:00'),
                                                               nvl(send_station_arrival_early_time, '0000-01-01 01:00:00')))) /
                                      3600
                             when next_station_network_code is null and deliver_early_scantime is null and
                                  sign_time is null
                                 then (unix_timestamp(concat('{{ execution_date | cst_ds }}', ' 23:59:59')) -
                                       unix_timestamp(greatest(nvl(send_station_send_later_time, '0000-01-01 01:00:00'),
                                                               nvl(send_station_arrival_early_time, '0000-01-01 01:00:00')))) /
                                      3600
                             when next_station_network_code is null and
                                  (deliver_early_scantime is not null or sign_time is not null)
                                 then (unix_timestamp(coalesce(deliver_early_scantime, sign_time)) -
                                       unix_timestamp(greatest(nvl(send_station_send_later_time, '0000-01-01 01:00:00'),
                                                               nvl(send_station_arrival_early_time, '0000-01-01 01:00:00')))) /
                                      3600
                             else 0 end            as duration
                       , send_station_network_code as duty_network_code
                       , send_station_network_name as duty_network_name
                       , timeout                   as duty_standard
                  from mate_date
                  where timeout is not null
              ) t
         where abs(duration) > duty_standard
     ),
     union_table as (
         select waybill_no
              , dt
              , first_duty_network_code
              , first_duty_network_name
              , first_duty_standard
              , first_duty_type
              , regexp_replace(concat_ws(',', collect_set(concat(duty_type, ':', duty_network_code))), '\\d+\:',
                               '')                                              as duty_network_code
              , regexp_replace(concat_ws(',',
                                         collect_set(concat(duty_type, ':', duty_network_name))), '\\d+\:',
                               '')                                              as duty_network_name
              , concat_ws(',', collect_set(concat(duty_type, ':', trace_falg))) as trace_falg
              , regexp_replace(concat_ws(',', collect_set(concat(duty_type, ':', duration))), '\\d+\:',
                               '')                                              as duration
              , regexp_replace(concat_ws(',', collect_set(concat(duty_type, ':', duty_standard))), '\\d+\:',
                               '')                                              as duty_standard
              , concat_ws(',', collect_set(duty_type))                          as duty_type
         from (
                  select *
                       , first_value(duty_network_code)
                                     over (partition by waybill_no order by duty_type,rn) as first_duty_network_code
                       , first_value(duty_network_name)
                                     over (partition by waybill_no order by duty_type,rn) as first_duty_network_name
                       , first_value(duty_standard)
                                     over (partition by waybill_no order by duty_type,rn) as first_duty_standard
                       , first_value(duty_type)
                                     over (partition by waybill_no order by duty_type,rn) as first_duty_type
                  from (
                           select *
--                          waybill_no,
--                          rn,
--                          dt,
--                          trace_falg,
--                          duration,
--                          duty_network_code,
--                          duty_network_name,
--                          duty_standard,
--                          duty_type
                           from first_type
                           where duty_network_code is not null
                           union all
                           select *
                           from second_type
                           where duty_network_code is not null
                           union all
                           select *
                           from third_type
                           where duty_network_code is not null
                           union all
                           select *
                           from four_type
                           where duty_network_code is not null
                       ) t
              ) t
         where first_duty_network_code is not null
         group by waybill_no
                , dt
                , first_duty_network_code
                , first_duty_network_name
                , first_duty_standard
                , first_duty_type
     )
insert
overwrite
table
jms_dm.dm_mng_thd_flase_send_waybill_detail_dt
partition
(
dt
)
select t1.waybill_no                                          as waybill_no              --运单号
     , t1.input_time                                          as input_time              --录入时间
     , substr(t1.input_time, 1, 10)                           as date_time               --日期
     , t2.first_duty_network_code                             as first_duty_network_code --首责任机构code
     , t2.first_duty_network_name                             as first_duty_network_name --首责任机构name
     , t4.network_type                                        as first_duty_network_type --首责任机构网点类型
     , t4.agent_code                                          as first_duty_agent_code   --首责任机构代理区code
     , t4.agent_name                                          as first_duty_agent_name   --首责任机构代理区name
     , t2.first_duty_type                                     as first_duty_type         --首责任类型
     , t2.first_duty_standard                                 as first_duty_standard     --首责任标准
     , t1.first_scan_time                                     as first_network_scantime  --首机构操作时间
     , t1.first_network_code                                  as first_network_code      --首机构操作网点
     , t1.first_network_name                                  as first_network_name      --首机构操作网点name
     , t1.first_network_agent_code                            as first_agent_code        --始发机构代理区code
     , t1.first_network_agent_name                            as first_agent_name        --始发机构代理区name
     , t1.first_network_city_id                               as first__city_id          --始发机构城市id
     , t1.first_network_city_desc                             as first__city_desc        --始发机构城市desc
     , t1.first_network_provider_id                           as first__provider_id      --始发机构省份id
     , t1.first_network_provider_desc                         as first__provider_desc    --始发机构省份desc
     , t1.dispatch_network_code                               as end_network_code        --派件网点
     , t1.dispatch_network_name                               as end_network_name        --派件网点name
     , t1.dispatch_network_agent_code                         as end_agent_code          --派件机构代理区code
     , t1.dispatch_network_agent_name                         as end_agent_name          --派件机构代理区name
     , t1.dispatch_network_city_id                            as end_city_id             --派件机构城市id
     , t1.dispatch_network_city_desc                          as end_city_desc           --派件机构城市desc
     , t1.dispatch_network_provider_id                        as end_province_id         --派件机构省份id
     , t1.dispatch_network_provider_desc                      as end_province_desc       --派件机构省份desc
     , t3.send_station_network_name                           as track_link              --链路环节name
     , t2.duty_network_code                                   as all_duty_network_code   --责任机构
     , t2.duty_network_name                                   as all_duty_network_name   --责任机构
     , t2.duration                                                                       --判责时间
     , t1.deliver_early_scansitecode                          as out_network_code        --出仓网点
     , t1.deliver_early_scansitename                          as out_network_name        --出仓网点
     , t1.deliver_city_id                                     as out_city_id             --出仓城市id
     , t1.deliver_city_desc                                   as out_city_desc           --出仓城市name
     , t1.deliver_provider_id                                 as out_province_id         --出仓省份id
     , t1.deliver_provider_desc                               as out_province_desc       --出仓省份name
     , t1.sign_time                                                                      --签收时间
     , t2.duty_standard                                       as all_duty_standard       --任标准
     , t2.duty_type                                           as all_duty_type           --责任类型
     , t2.trace_falg                                          as all_trace_falg          --问题环节
     , t3.send_station_network_code                           as track_link_code         --链路环节code
     , t5.upload_batch                                        as batch_no                --批次号
     , if(t5.bill_no is not null, 1, 0)                       as is_manual               --是否手工上传
     , from_unixtime(unix_timestamp(), 'yyyy-MM-dd HH:mm:ss') as update_time
     , t1.dt
from mate_date t1
         join union_table t2
              on t1.waybill_no = t2.waybill_no
         join line_info t3
              on t1.waybill_no = t3.waybill_no
         left join jms_dim.dim_network_whole_massage t4
                   on t2.first_duty_network_code = t4.code
         left join ship_upload_bill_tab t5
                   on t1.waybill_no = t5.bill_no
where t1.rn = 1
    distribute by dt
    , pmod(hash(waybill_no), 100)
;